package com.cetccloud.ap.proxy.extractor;

import cn.hutool.core.map.MapUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.cetccloud.ap.common.RuntimeCache;
import com.cetccloud.ap.config.ProxyConfig;
import com.cetccloud.ap.mapper.HttpSubscribeMapper;
import com.cetccloud.ap.mapper.pojo.HttpSubscribe;
import com.cetccloud.ap.proxy.channel.Channel;
import com.cetccloud.ap.proxy.processor.subscribe.model.SubscribeHttpParam;
import com.cetccloud.ap.util.KafkaKeyUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.ExchangeStrategies;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @ClassName SubscribeHttpExtractor
 * @Description TODO
 * @Author superz
 * @Date 2021/4/21 17:37
 * @Version 1.0
 **/
@Slf4j
@Component
public class SubscribeHttpExtractor {

    @Autowired
    private ProxyConfig proxyConfig;

    @Autowired
    private Channel crossDomainChannel;

    @Autowired
    private HttpSubscribeMapper httpSubscribeMapper;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private RuntimeCache runtimeCache;

//    @Scheduled(cron = "0 */5 * * * ?")
    public void scheduled() {
        log.info("### SubscribeHttpExtractor scheduled ###");
        List<HttpSubscribe> list = httpSubscribeMapper.selectList(Wrappers.query());

        for (HttpSubscribe httpSubscribe : list) {
            log.info("### deal ### {}", JSON.toJSONString(httpSubscribe));

            Long currIncrement = httpSubscribe.getCurrIncrement();
            SubscribeHttpParam.Increment increment = JSON.parseObject(httpSubscribe.getIncrement(), SubscribeHttpParam.Increment.class);
            String incrementField = increment.getField();

            JSONObject body = JSON.parseObject(httpSubscribe.getBody());
            JSONArray filter = body.getJSONArray("filter");
            // 设置当前增量字段值
            for (Object f : filter) {
                String filterField = ((JSONObject) f).getString("filterField");
                if (StringUtils.equals(incrementField, filterField)) {
                    ((JSONObject) f).put("value", null != currIncrement ? increment.getInit() : currIncrement);
                    break;
                }
            }

            // 调用s10接口查询数据
            // 除了使用create快速生成 外还可以使用builder创建
            Mono<String> mono = WebClient.builder()
                    .exchangeStrategies(ExchangeStrategies.builder()
                            .codecs(clientCodecConfigurer -> clientCodecConfigurer.
                                    defaultCodecs().maxInMemorySize(10*1024*1024))
                            .build())
                    .baseUrl(httpSubscribe.getUrl()).build()
                    // 将info body 写入 proxy请求的body中
                    .method(HttpMethod.resolve(StringUtils.upperCase(httpSubscribe.getMethod())))
                    .headers(httpHeaders -> {
                        Map<String, String> headers = JSON.parseObject(httpSubscribe.getHeader(), Map.class);
                        if (MapUtil.isNotEmpty(headers)) {
                            headers.entrySet().stream().forEach(entry -> httpHeaders.set(entry.getKey(), entry.getValue()));
                        }
                        // 设置s10的token头
                        if (StringUtils.isNotBlank(runtimeCache.getS10Token())) {
                            headers.put("token", runtimeCache.getS10Token());
                        }
                    })
                    .contentType(MediaType.APPLICATION_JSON).bodyValue(body)
                    .retrieve()
                    .bodyToMono(String.class)
                    .timeout(Duration.ofMillis(30_000));

            mono.subscribe(v -> {
                log.info("### deal ### success: {}", v);
                JSONObject result = JSON.parseObject(v);
                String resultCode = result.getString("resultCode");
                // 1000: 成功，2001: 服务器错误，2003: 参数错误
                if ("1000".equals(resultCode)) {
                    JSONObject data = result.getJSONObject("data");
                    JSONArray fieldsValue = data.getJSONArray("fieldsValue");
                    if (CollectionUtils.isNotEmpty(fieldsValue)) {
                        long maxIncrement = Long.valueOf(currIncrement);
                        for (Object value : fieldsValue) {
                            long index = ((JSONObject) value).getLongValue(incrementField);
                            if (index > maxIncrement) {
                                maxIncrement = index;
                            }
                            // CallTypeEnum.KAFKA_TRANSFER同步逻辑
                            String topicPrefix = proxyConfig.getKafka().getPrefix();
                            ProducerRecord<String, String> producerRecord =
                                    new ProducerRecord(topicPrefix + httpSubscribe.getTopic(), KafkaKeyUtil.zoneByKey(httpSubscribe.getZoneId()), JSON.toJSONString(value));
                            kafkaTemplate.send(producerRecord).addCallback(
                                    success -> {
                                        String topic = success.getRecordMetadata().topic();
                                        int partition = success.getRecordMetadata().partition();
                                        long offset = success.getRecordMetadata().offset();
                                        log.debug("发送消息成功: {}, {}, {}", topic, partition, offset);
                                    },
                                    failure -> {
                                        log.error("发送消息失败: " + failure.getMessage());
                                    });
                        }
                        // 更新tb_http_subscribe.curr_increment
                        httpSubscribeMapper.updateCurrIncrement(httpSubscribe.getId(), maxIncrement);
                    }
                }
            }, t -> {
                log.error("### deal ### failed", t);
            });
        }
    }
}
